#!/usr/bin/env python
#==============================================================================
#
#  $Id$
#
"""
    Simple tests of the "full" implementation.  See also test_1000.
"""
#
#   Copyright (C) 2006 Michael A. Muller
#
#   This file is part of ODB.
#
#   ODB is free software: you can redistribute it and/or modify it under the 
#   terms of the GNU Lesser General Public License as published by the Free 
#   Software Foundation, either version 3 of the License, or (at your option) 
#   any later version.
#
#   ODB is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU Lesser General Public License for more details.
#
#   You should have received a copy of the GNU Lesser General Public License 
#   along with ODB.  If not, see <http://www.gnu.org/licenses/>.
#
#==============================================================================

import shutil, unittest, string
import fixpath
from odb.full import FullEnv
from utils import DBTest

class test_full(DBTest):

    def testImmediateCheckpoint(self):
        otherDB = self.env.getBTreeDB('other')
        for key, val in zip([`i` for i in range(1, 10)], range(1, 10)):
            self.db[key] = val
        self.env.checkpoint()
        self.reload()
        
        # force a demand load of the root by getting the first element
        self.db.cursor().next()
        
        # checkpoint - this will necessarily involve the checkpointing of stub 
        # nodes.
        self.env.checkpoint()
    
    def testLoadOnInit(self):
        # checkpoint to start a new log
        self.env.checkpoint()
        
        # put an insert action into the new log and close the environment
        self.db['first'] = 100
        
        # force reading the new log, verifying that the table gets loaded even 
        # though the first reference to it is in an insert transaction
        self.reload()
    
    def testRedistribute(self):
        # add four items to the database, reload and start deleting them.  
        # This forces a redistribute involving an unloaded node
        
        for key, val in zip('abcd', range(4)):
            self.db[key] = val
        self.env.checkpoint()

        self.reload()
        txn = self.env.getTxn()
        self.db.delete(txn, 'a')
#        for key in string.lowercase:
#            self.db.delete(txn, key)
        txn.commit()
    
    def testReverseCursorDemandLoads(self):
        items = [('a', 100), ('b', 200), ('c', 300), 
                 ('d', 400)
                 ]
        for key, val in items:
            self.db[key] = val
        self.env.checkpoint()
        self.reload()
        
        cur = self.db.cursor()
        items.reverse()
        for key, val in items:
            self.assertEquals(cur.prev(), (key, val))
        self.assertRaises(StopIteration, cur.prev)
    
    def testNextOnEmpty(self):
        # tests a bug condition I discovered with doing a next after a 
        # setToKey()
        cur = self.db.cursor()
        cur.setToKey('foo')
        self.assertRaises(StopIteration, cur.next)
    
    def testEmptyKeyAfterCheckpoint(self):
        # verify that an empty key is not converted to None after a checkpoint.
        self.db[''] = 100
        self.env.checkpoint()
        self.reload()
        self.assertRaises(KeyError, lambda db = self.db: db[None])
        self.assertEquals(self.db[''], 100)

    def testNoDupsOnSet(self):
        txn = self.env.getTxn()
        self.db.set(txn, 'a', 100)
        self.db.set(txn, 'a', 200)
        txn.commit()
        self.reload()
        self.assertEquals(len([x for x in self.db.cursor()]), 1)
        
    # XXX need to test the "short circuit to the end of the list" optimization 
    # in BTreeMeat.find()

class TestBList(unittest.TestCase):

    def reload(self):
        self.env = FullEnv('odb_test_db')
        self.db = self.env.getBListDB('blist')
        return self.env, self.db

    def setUp(self):
        self.reload()
        self.refList = []
    
    def tearDown(self):
        del self.env, self.db
	shutil.rmtree('odb_test_db')

    def append(self, item, txn = None):
        self.db.append(item, txn)
        self.refList.append(item)

    def check(self):
        for refVal, dbVal in zip(self.refList, self.db.cursor()):
            self.assertEquals(refVal, dbVal)

    def testAppends(self):
        txn = self.db.getTxn()
        for i in range(10):
            self.append(i, txn)
        txn.commit()
        self.check()
    
    def testLen(self):
        for i in range(7):
            self.append(i)
        self.assertEquals(len(self.db), i + 1)
        txn = self.db.getTxn()
        self.assertEquals(self.db.len(txn), i + 1)
        txn.commit()
        self.assertRaises(AssertionError, self.db.len, txn)

unittest.main()